/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server.quorum;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.zookeeper.jmx.MBeanRegistry;
import org.apache.zookeeper.server.ZooKeeperThread;
import org.apache.zookeeper.server.quorum.QuorumCnxManager.Message;
import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Implementation of leader election using TCP. It uses an object of the class
 * QuorumCnxManager to manage connections. Otherwise, the algorithm is
 * push-based as with the other UDP implementations.
 *
 * There are a few parameters that can be tuned to change its behavior. First,
 * finalizeWait determines the amount of time to wait until deciding upon a
 * leader. This is part of the leader election algorithm.
 */

public class FastLeaderElection implements Election {
	private static final Logger LOG = LoggerFactory.getLogger(FastLeaderElection.class);

	/**
	 * Determine how much time a process has to wait once it believes that it
	 * has reached the end of leader election.
	 */
	final static int finalizeWait = 200;

	/**
	 * Upper bound on the amount of time between two consecutive notification
	 * checks. This impacts the amount of time to get the system up again after
	 * long partitions. Currently 60 seconds.
	 */

	final static int maxNotificationInterval = 60000;

	/**
	 * Connection manager. Fast leader election uses TCP for communication
	 * between peers, and QuorumCnxManager manages such connections.
	 */

	QuorumCnxManager manager;

	/**
	 * Notifications are messages that let other peers know that a given peer
	 * has changed its vote, either because it has joined leader election or
	 * because it learned of another peer with higher zxid or same zxid and
	 * higher server id
	 */

	static public class Notification {
		/*
		 * Format version, introduced in 3.4.6
		 */

		public final static int CURRENTVERSION = 0x1;
		int version;

		/*
		 * Proposed leader
		 */
		long leader;

		/*
		 * zxid of the proposed leader
		 */
		long zxid;

		/*
		 * Epoch
		 */
		long electionEpoch;

		/*
		 * current state of sender
		 */
		QuorumPeer.ServerState state;

		/*
		 * Address of sender
		 */
		long sid;

		/*
		 * epoch of the proposed leader
		 */
		long peerEpoch;

		@Override
		public String toString() {
			return new String(Long.toHexString(version) + " (message format version), " + leader + " (n.leader), 0x"
					+ Long.toHexString(zxid) + " (n.zxid), 0x" + Long.toHexString(electionEpoch) + " (n.round), "
					+ state + " (n.state), " + sid + " (n.sid), 0x" + Long.toHexString(peerEpoch) + " (n.peerEpoch) ");
		}
	}

	static ByteBuffer buildMsg(int state, long leader, long zxid, long electionEpoch, long epoch) {
		byte requestBytes[] = new byte[40];
		ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);

		/*
		 * Building notification packet to send
		 */

		requestBuffer.clear();
		requestBuffer.putInt(state);
		requestBuffer.putLong(leader);
		requestBuffer.putLong(zxid);
		requestBuffer.putLong(electionEpoch);
		requestBuffer.putLong(epoch);
		requestBuffer.putInt(Notification.CURRENTVERSION);

		return requestBuffer;
	}

	/**
	 * Messages that a peer wants to send to other peers. These messages can be
	 * both Notifications and Acks of reception of notification.
	 */
	static public class ToSend {
		static enum mType {
			crequest, challenge, notification, ack
		}

		ToSend(mType type, long leader, long zxid, long electionEpoch, ServerState state, long sid, long peerEpoch) {

			this.leader = leader;
			this.zxid = zxid;
			this.electionEpoch = electionEpoch;
			this.state = state;
			this.sid = sid;
			this.peerEpoch = peerEpoch;
		}

		/*
		 * Proposed leader in the case of notification
		 */
		long leader;

		/*
		 * id contains the tag for acks, and zxid for notifications
		 */
		long zxid;

		/*
		 * Epoch
		 */
		long electionEpoch;

		/*
		 * Current state;
		 */
		QuorumPeer.ServerState state;

		/*
		 * Address of recipient
		 */
		long sid;

		/*
		 * Leader epoch
		 */
		long peerEpoch;
	}

	LinkedBlockingQueue<ToSend> sendqueue;
	LinkedBlockingQueue<Notification> recvqueue;

	/**
	 * Multi-threaded implementation of message handler. Messenger implements
	 * two sub-classes: WorkReceiver and WorkSender. The functionality of each
	 * is obvious from the name. Each of these spawns a new thread.
	 */

	protected class Messenger {

		/**
		 * Receives messages from instance of QuorumCnxManager on method run(),
		 * and processes such messages.
		 */

		class WorkerReceiver extends ZooKeeperThread {
			volatile boolean stop;
			QuorumCnxManager manager;

			WorkerReceiver(QuorumCnxManager manager) {
				super("WorkerReceiver");
				this.stop = false;
				this.manager = manager;
			}

			public void run() {

				Message response;
				while (!stop) {
					// Sleeps on receive
					try {
						response = manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
						if (response == null)
							continue;

						/*
						 * If it is from an observer, respond right away. Note
						 * that the following predicate assumes that if a server
						 * is not a follower, then it must be an observer. If we
						 * ever have any other type of learner in the future,
						 * we'll have to change the way we check for observers.
						 */
						if (!self.getVotingView().containsKey(response.sid)) {
							Vote current = self.getCurrentVote();
							ToSend notmsg = new ToSend(ToSend.mType.notification, current.getId(), current.getZxid(),
									logicalclock, self.getPeerState(), response.sid, current.getPeerEpoch());

							sendqueue.offer(notmsg);
						} else {
							// Receive new message
							if (LOG.isDebugEnabled()) {
								LOG.debug("Receive new notification message. My id = " + self.getId());
							}

							/*
							 * We check for 28 bytes for backward compatibility
							 */
							if (response.buffer.capacity() < 28) {
								LOG.error("Got a short response: " + response.buffer.capacity());
								continue;
							}
							boolean backCompatibility = (response.buffer.capacity() == 28);
							response.buffer.clear();

							// Instantiate Notification and set its attributes
							Notification n = new Notification();

							// State of peer that sent this message
							QuorumPeer.ServerState ackstate = QuorumPeer.ServerState.LOOKING;
							switch (response.buffer.getInt()) {
							case 0:
								ackstate = QuorumPeer.ServerState.LOOKING;
								break;
							case 1:
								ackstate = QuorumPeer.ServerState.FOLLOWING;
								break;
							case 2:
								ackstate = QuorumPeer.ServerState.LEADING;
								break;
							case 3:
								ackstate = QuorumPeer.ServerState.OBSERVING;
								break;
							default:
								continue;
							}

							n.leader = response.buffer.getLong();
							n.zxid = response.buffer.getLong();
							n.electionEpoch = response.buffer.getLong();
							n.state = ackstate;
							n.sid = response.sid;
							if (!backCompatibility) {
								n.peerEpoch = response.buffer.getLong();
							} else {
								if (LOG.isInfoEnabled()) {
									LOG.info("Backward compatibility mode, server id=" + n.sid);
								}
								n.peerEpoch = ZxidUtils.getEpochFromZxid(n.zxid);
							}

							/*
							 * Version added in 3.4.6
							 */

							n.version = (response.buffer.remaining() >= 4) ? response.buffer.getInt() : 0x0;

							/*
							 * Print notification info
							 */
							if (LOG.isInfoEnabled()) {
								printNotification(n);
							}

							/*
							 * If this server is looking, then send proposed
							 * leader
							 */

							if (self.getPeerState() == QuorumPeer.ServerState.LOOKING) {
								recvqueue.offer(n);

								/*
								 * Send a notification back if the peer that
								 * sent this message is also looking and its
								 * logical clock is lagging behind.
								 */
								if ((ackstate == QuorumPeer.ServerState.LOOKING) && (n.electionEpoch < logicalclock)) {
									Vote v = getVote();
									ToSend notmsg = new ToSend(ToSend.mType.notification, v.getId(), v.getZxid(),
											logicalclock, self.getPeerState(), response.sid, v.getPeerEpoch());
									sendqueue.offer(notmsg);
								}
							} else {
								/*
								 * If this server is not looking, but the one
								 * that sent the ack is looking, then send back
								 * what it believes to be the leader.
								 */
								Vote current = self.getCurrentVote();
								if (ackstate == QuorumPeer.ServerState.LOOKING) {
									if (LOG.isDebugEnabled()) {
										LOG.debug("Sending new notification. My id =  " + self.getId() + " recipient="
												+ response.sid + " zxid=0x" + Long.toHexString(current.getZxid())
												+ " leader=" + current.getId());
									}

									ToSend notmsg;
									if (n.version > 0x0) {
										notmsg = new ToSend(ToSend.mType.notification, current.getId(),
												current.getZxid(), current.getElectionEpoch(), self.getPeerState(),
												response.sid, current.getPeerEpoch());

									} else {
										Vote bcVote = self.getBCVote();
										notmsg = new ToSend(ToSend.mType.notification, bcVote.getId(), bcVote.getZxid(),
												bcVote.getElectionEpoch(), self.getPeerState(), response.sid,
												bcVote.getPeerEpoch());
									}
									sendqueue.offer(notmsg);
								}
							}
						}
					} catch (InterruptedException e) {
						System.out.println("Interrupted Exception while waiting for new message" + e.toString());
					}
				}
				LOG.info("WorkerReceiver is down");
			}
		}

		/**
		 * This worker simply dequeues a message to send and and queues it on
		 * the manager's queue.
		 */

		class WorkerSender extends ZooKeeperThread {
			volatile boolean stop;
			QuorumCnxManager manager;

			WorkerSender(QuorumCnxManager manager) {
				super("WorkerSender");
				this.stop = false;
				this.manager = manager;
			}

			public void run() {
				while (!stop) {
					try {
						ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
						if (m == null)
							continue;

						process(m);
					} catch (InterruptedException e) {
						break;
					}
				}
				LOG.info("WorkerSender is down");
			}

			/**
			 * Called by run() once there is a new message to send.
			 *
			 * @param m
			 *            message to send
			 */
			void process(ToSend m) {
				ByteBuffer requestBuffer = buildMsg(m.state.ordinal(), m.leader, m.zxid, m.electionEpoch, m.peerEpoch);
				manager.toSend(m.sid, requestBuffer);
			}
		}

		/**
		 * Test if both send and receive queues are empty.
		 */
		public boolean queueEmpty() {
			return (sendqueue.isEmpty() || recvqueue.isEmpty());
		}

		WorkerSender ws;
		WorkerReceiver wr;

		/**
		 * Constructor of class Messenger.
		 *
		 * @param manager
		 *            Connection manager
		 */
		Messenger(QuorumCnxManager manager) {

			this.ws = new WorkerSender(manager);

			Thread t = new Thread(this.ws, "WorkerSender[myid=" + self.getId() + "]");
			t.setDaemon(true);
			t.start();

			this.wr = new WorkerReceiver(manager);

			t = new Thread(this.wr, "WorkerReceiver[myid=" + self.getId() + "]");
			t.setDaemon(true);
			t.start();
		}

		/**
		 * Stops instances of WorkerSender and WorkerReceiver
		 */
		void halt() {
			this.ws.stop = true;
			this.wr.stop = true;
		}

	}

	QuorumPeer self;
	Messenger messenger;
	/**
	 * 逻辑时钟
                     或者叫投票的次数，同一轮投票过程中的逻辑时钟值是相同的。每投完一次票这个数据就会增加，
                     然后与接收到的其它服务器返回的投票信息中的数值相比，根据不同的值做出不同的判断
	 */
	volatile long logicalclock; /* Election instance */
	//提案投票的Leader
	long proposedLeader;
	long proposedZxid;
	long proposedEpoch;

	/**
	 * Returns the current vlue of the logical clock counter
	 */
	public long getLogicalClock() {
		return logicalclock;
	}

	/**
	 * Constructor of FastLeaderElection. It takes two parameters, one is the
	 * QuorumPeer object that instantiated this object, and the other is the
	 * connection manager. Such an object should be created only once by each
	 * peer during an instance of the ZooKeeper service.
	 *
	 * @param self
	 *            QuorumPeer that created this object
	 * @param manager
	 *            Connection manager
	 */
	public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager) {
		this.stop = false;
		this.manager = manager;
		starter(self, manager);
	}

	/**
	 * This method is invoked by the constructor. Because it is a part of the
	 * starting procedure of the object that must be on any constructor of this
	 * class, it is probably best to keep as a separate method. As we have a
	 * single constructor currently, it is not strictly necessary to have it
	 * separate.
	 *
	 * @param self
	 *            QuorumPeer that created this object
	 * @param manager
	 *            Connection manager
	 */
	private void starter(QuorumPeer self, QuorumCnxManager manager) {
		this.self = self;
		proposedLeader = -1;
		proposedZxid = -1;

		sendqueue = new LinkedBlockingQueue<ToSend>();
		recvqueue = new LinkedBlockingQueue<Notification>();
		this.messenger = new Messenger(manager);
	}

	private void leaveInstance(Vote v) {
		if (LOG.isDebugEnabled()) {
			LOG.debug("About to leave FLE instance: leader=" + v.getId() + ", zxid=0x" + Long.toHexString(v.getZxid())
					+ ", my id=" + self.getId() + ", my state=" + self.getPeerState());
		}
		recvqueue.clear();
	}

	public QuorumCnxManager getCnxManager() {
		return manager;
	}

	volatile boolean stop;

	public void shutdown() {
		stop = true;
		LOG.debug("Shutting down connection manager");
		manager.halt();
		LOG.debug("Shutting down messenger");
		messenger.halt();
		LOG.debug("FLE is down");
	}

	/**
	 * Send notifications to all peers upon a change in our vote
	 */
	private void sendNotifications() {
		for (QuorumServer server : self.getVotingView().values()) {
			long sid = server.id;

			ToSend notmsg = new ToSend(ToSend.mType.notification, proposedLeader, proposedZxid, logicalclock,
					QuorumPeer.ServerState.LOOKING, sid, proposedEpoch);
			if (LOG.isDebugEnabled()) {
				LOG.debug("Sending Notification: " + proposedLeader + " (n.leader), 0x" + Long.toHexString(proposedZxid)
						+ " (n.zxid), 0x" + Long.toHexString(logicalclock) + " (n.round), " + sid + " (recipient), "
						+ self.getId() + " (myid), 0x" + Long.toHexString(proposedEpoch) + " (n.peerEpoch)");
			}
			sendqueue.offer(notmsg);
		}
	}

	private void printNotification(Notification n) {
		LOG.info("Notification: " + n.toString() + self.getPeerState() + " (my state)");
	}

	/**
	 * Check if a pair (server id, zxid) succeeds our current vote.
	 *
	 * @param id
	 *            Server identifier
	 * @param zxid
	 *            Last zxid observed by the issuer of this vote
	 */
	protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid,
			long curEpoch) {
		LOG.debug("id: " + newId + ", proposed id: " + curId + ", zxid: 0x" + Long.toHexString(newZxid)
				+ ", proposed zxid: 0x" + Long.toHexString(curZxid));
		if (self.getQuorumVerifier().getWeight(newId) == 0) {
			return false;
		}

		/*
		 * We return true if one of the following three cases hold: 1- New epoch
		 * is higher 2- New epoch is the same as current epoch, but new zxid is
		 * higher 3- New epoch is the same as current epoch, new zxid is the
		 * same as current zxid, but server id is higher.
		 */

		return ((newEpoch > curEpoch)
				|| ((newEpoch == curEpoch) && ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))));
	}

	/**
	 * Termination predicate. Given a set of votes, determines if have
	 * sufficient to declare the end of the election round.
	 *
	 * @param votes
	 *            Set of votes
	 * @param l
	 *            Identifier of the vote received last
	 * @param zxid
	 *            zxid of the the vote received last
	 */
	protected boolean termPredicate(HashMap<Long, Vote> votes, Vote vote) {

		HashSet<Long> set = new HashSet<Long>();

		/*
		 * First make the views consistent. Sometimes peers will have different
		 * zxids for a server depending on timing.
		 */
		for (Map.Entry<Long, Vote> entry : votes.entrySet()) {
			if (vote.equals(entry.getValue())) {
				set.add(entry.getKey());
			}
		}

		return self.getQuorumVerifier().containsQuorum(set);
	}

	/**
	 * In the case there is a leader elected, and a quorum supporting this
	 * leader, we have to check if the leader has voted and acked that it is
	 * leading. We need this check to avoid that peers keep electing over and
	 * over a peer that has crashed and it is no longer leading.
	 *
	 * @param votes
	 *            set of votes
	 * @param leader
	 *            leader id
	 * @param electionEpoch
	 *            epoch id
	 */
	protected boolean checkLeader(HashMap<Long, Vote> votes, long leader, long electionEpoch) {

		boolean predicate = true;

		/*
		 * If everyone else thinks I'm the leader, I must be the leader. The
		 * other two checks are just for the case in which I'm not the leader.
		 * If I'm not the leader and I haven't received a message from leader
		 * stating that it is leading, then predicate is false.
		 */

		if (leader != self.getId()) {
			if (votes.get(leader) == null)
				predicate = false;
			else if (votes.get(leader).getState() != ServerState.LEADING)
				predicate = false;
		} else if (logicalclock != electionEpoch) {
			predicate = false;
		}

		return predicate;
	}

	/**
	 * This predicate checks that a leader has been elected. It doesn't make a
	 * lot of sense without context (check lookForLeader) and it has been
	 * separated for testing purposes.
	 * 
	 * @param recv
	 *            map of received votes
	 * @param ooe
	 *            map containing out of election votes (LEADING or FOLLOWING)
	 * @param n
	 *            Notification
	 * @return
	 */
	protected boolean ooePredicate(HashMap<Long, Vote> recv, HashMap<Long, Vote> ooe, Notification n) {

		return (termPredicate(recv, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state))
				&& checkLeader(ooe, n.leader, n.electionEpoch));

	}

	synchronized void updateProposal(long leader, long zxid, long epoch) {
		if (LOG.isDebugEnabled()) {
			LOG.debug("Updating proposal: " + leader + " (newleader), 0x" + Long.toHexString(zxid) + " (newzxid), "
					+ proposedLeader + " (oldleader), 0x" + Long.toHexString(proposedZxid) + " (oldzxid)");
		}
		proposedLeader = leader;
		proposedZxid = zxid;
		proposedEpoch = epoch;
	}

	synchronized Vote getVote() {
		return new Vote(proposedLeader, proposedZxid, proposedEpoch);
	}

	/**
	 * A learning state can be either FOLLOWING or OBSERVING. This method simply
	 * decides which one depending on the role of the server.
	 *
	 * @return ServerState
	 */
	private ServerState learningState() {
		if (self.getLearnerType() == LearnerType.PARTICIPANT) {
			LOG.debug("I'm a participant: " + self.getId());
			return ServerState.FOLLOWING;
		} else {
			LOG.debug("I'm an observer: " + self.getId());
			return ServerState.OBSERVING;
		}
	}

	/**
	 * Returns the initial vote value of server identifier.
	 *
	 * @return long
	 */
	private long getInitId() {
		if (self.getLearnerType() == LearnerType.PARTICIPANT)
			return self.getId();
		else
			return Long.MIN_VALUE;
	}

	/**
	 * Returns initial last logged zxid.
	 *
	 * @return long
	 */
	private long getInitLastLoggedZxid() {
		if (self.getLearnerType() == LearnerType.PARTICIPANT)
			return self.getLastLoggedZxid();
		else
			return Long.MIN_VALUE;
	}

	/**
	 * Returns the initial vote value of the peer epoch.
	 *
	 * @return long
	 */
	private long getPeerEpoch() {
		if (self.getLearnerType() == LearnerType.PARTICIPANT)
			try {
				return self.getCurrentEpoch();
			} catch (IOException e) {
				RuntimeException re = new RuntimeException(e.getMessage());
				re.setStackTrace(e.getStackTrace());
				throw re;
			}
		else
			return Long.MIN_VALUE;
	}

	/**
	 * Starts a new round of leader election. Whenever our QuorumPeer changes
	 * its state to LOOKING, this method is invoked, and it sends notifications
	 * to all other peers.
	 * http://www.th7.cn/Program/java/201702/1115229.shtml
	 */
	public Vote lookForLeader() throws InterruptedException {
		try {
			self.jmxLeaderElectionBean = new LeaderElectionBean();
			MBeanRegistry.getInstance().register(self.jmxLeaderElectionBean, self.jmxLocalPeerBean);
		} catch (Exception e) {
			LOG.warn("Failed to register with JMX", e);
			self.jmxLeaderElectionBean = null;
		}
		if (self.start_fle == 0) {
			self.start_fle = System.currentTimeMillis();
		}
		try {
			HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();

			HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();

			int notTimeout = finalizeWait;

			synchronized (this) {				
				logicalclock++;
				//给自己投票
				updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
			}

			LOG.info("New election. My id =  " + self.getId() + ", proposed zxid=0x" + Long.toHexString(proposedZxid));
			//将ToSendMsg放入sendqueue队列中;
			sendNotifications();

			/*
			 * 循环，如果是竞选状态一直到选举出结果
			 * Loop in which we exchange notifications until we find a leader
			 */

			while ((self.getPeerState() == ServerState.LOOKING) && (!stop)) {
				/*
				 * Remove next notification from queue, times out after 2 times
				 * the termination time
				 */
				//如果在notTimeout毫秒内成功地移除了队列头元素，则立即返回头元素；否则在到达超时时，返回null。
				Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);

				/*
				 * Sends more notifications if haven't received enough.
				 * Otherwise processes new notification.
				 * 没有收到投票信息
				 */
				if (n == null) {
					if (manager.haveDelivered()) {
						sendNotifications();
					} else {
						manager.connectAll();
					}

					/* 
					 * http://www.tuicool.com/articles/zayq22
					 * Exponential backoff【指数退避算法】
					 * 退避算法就是网络上的节点在发送数据冲突后，等待一定时间后再发，等待时间是随指数增长，从而避免频繁的触发冲突。
					 */
					int tmpTimeOut = notTimeout * 2;
					notTimeout = (tmpTimeOut < maxNotificationInterval ? tmpTimeOut : maxNotificationInterval);
					LOG.info("Notification time out: " + notTimeout);
				} else if (self.getVotingView().containsKey(n.sid)) {
					/*
					 * Only proceed if the vote comes from a replica in the
					 * voting view.
					 */
					switch (n.state) {
					case LOOKING:
						// If notification > current, replace and send messages
						// out
						// 根据Epoch判断投票是否超前,或者过时;或者是当前轮次
						if (n.electionEpoch > logicalclock) {
							logicalclock = n.electionEpoch;
							recvset.clear();
							//更新投票信息(如果接受的别的server的Notification的Vote更大,则用其替代当前的Vote)
							if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, getInitId(), getInitLastLoggedZxid(),
									getPeerEpoch())) {
								updateProposal(n.leader, n.zxid, n.peerEpoch);
							} else {
								updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
							}
							//发送投票信息
							sendNotifications();
						} else if (n.electionEpoch < logicalclock) {
							if (LOG.isDebugEnabled()) {
								LOG.debug(
										"Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x"
												+ Long.toHexString(n.electionEpoch) + ", logicalclock=0x"
												+ Long.toHexString(logicalclock));
							}
							break;
						} else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid,
								proposedEpoch)) {//look at Line 763 for init
							//更新投票信息
							updateProposal(n.leader, n.zxid, n.peerEpoch);
							sendNotifications();
						}

						if (LOG.isDebugEnabled()) {
							LOG.debug("Adding vote: from=" + n.sid + ", proposed leader=" + n.leader
									+ ", proposed zxid=0x" + Long.toHexString(n.zxid) + ", proposed election epoch=0x"
									+ Long.toHexString(n.electionEpoch));
						}

						recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
						//判断是否投票结束
						if (termPredicate(recvset,
								new Vote(proposedLeader, proposedZxid, logicalclock, proposedEpoch))) {

							// Verify if there is any change in the proposed leader
							// 如果 n从recvqueue取出!=null,并且进入Line：856；则重新走Line：775的逻辑
							while ((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null) {
								if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid,
										proposedEpoch)) {
									recvqueue.put(n);
									break;
								}
							}

							/*
							 * This predicate is true once we don't read any new
							 * relevant message from the reception queue
							 */
							if (n == null) {
							    //如果提案选举出的LeaderId等于自身sid,当前server角色就是Leader
								self.setPeerState(
										(proposedLeader == self.getId()) ? ServerState.LEADING : learningState());

								Vote endVote = new Vote(proposedLeader, proposedZxid, logicalclock, proposedEpoch);
								//recvqueue.clear();
								leaveInstance(endVote);
								return endVote;
							}
						}
						break;
					case OBSERVING:
						LOG.debug("Notification from observer: " + n.sid);
						break;
					case FOLLOWING:
					case LEADING:
						/*
						 * Consider all notifications from the same epoch
						 * together.
						 */
						if (n.electionEpoch == logicalclock) {
							recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));

							if (ooePredicate(recvset, outofelection, n)) {
								self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());

								Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
								leaveInstance(endVote);
								return endVote;
							}
						}

						/*
						 * Before joining an established ensemble, verify a
						 * majority is following the same leader.
						 */
						//记录投票已经完成
						outofelection.put(n.sid,
								new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));

						if (ooePredicate(outofelection, outofelection, n)) {
							synchronized (this) {
								logicalclock = n.electionEpoch;
								self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());
							}
							Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
							leaveInstance(endVote);
							return endVote;
						}
						break;
					default:
						LOG.warn("Notification state unrecognized: {} (n.state), {} (n.sid)", n.state, n.sid);
						break;
					}
				} else {
					LOG.warn("Ignoring notification from non-cluster member " + n.sid);
				}
			}
			return null;
		} finally {
			try {
				if (self.jmxLeaderElectionBean != null) {
					MBeanRegistry.getInstance().unregister(self.jmxLeaderElectionBean);
				}
			} catch (Exception e) {
				LOG.warn("Failed to unregister with JMX", e);
			}
			self.jmxLeaderElectionBean = null;
		}
	}
}
